package com.strato.batch.base.v1_0_0.batch;

import com.strato.batch.base.v1_0_0.api.Job;
import com.strato.batch.base.v1_0_0.api.JobConsumer;
import com.strato.batch.base.v1_0_0.api.JobProducer;
import com.strato.logging.v1_0_0.api.LogFactory;
import com.strato.transaction.v1_0_0.api.TransactionStatusHolder;
import com.strato.transaction.v1_0_0.api.jdbc.DataSourceTransactionStatus;

public class ExecutorNode<T> implements Runnable {

	private JobProducer<T> jobProvider;

	private BatchJobExecutionConfig<T> config;

	public ExecutorNode(JobProducer<T> jobProvider, BatchJobExecutionConfig<T> config) {
		super();
		this.jobProvider = jobProvider;
		this.config = config;
	}

	private boolean running = true;

	@Override
	public void run() {
		while (running) {
				if (jobProvider.available()) {
					DataSourceTransactionStatus qTX = new DataSourceTransactionStatus();
					TransactionStatusHolder.hold(qTX);
					Job<T> job=null;
					try{
						job = jobProvider.produce();
						qTX.commit();
					}catch(Exception e){
						qTX.rollback();
						LogFactory.getLog(getClass()).error("Consumer produce error", e);
//						throw new RuntimeException(e);
					}finally{
						qTX.close();
						TransactionStatusHolder.unhold();
					}
					if (job == null) {
						try {
							Thread.sleep(2000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					} else {
						DataSourceTransactionStatus txStatus = new DataSourceTransactionStatus();
						TransactionStatusHolder.hold(txStatus);
						try {
							doJob(job);
							txStatus.commit();
						} catch (Exception e) {
							e.printStackTrace();
							txStatus.rollback();
							LogFactory.getLog(getClass()).error("Consumer consume error", e);
//							throw new RuntimeException(e);
						} finally {
							txStatus.close();
							TransactionStatusHolder.unhold();
						}
					}
				} else {
					stop();
				}
		}
	}

	private void doJob(Job<T> job) {
		JobConsumer<T> consumer = config.getJobConsumerFactory().obtainConsumer();
		consumer.consume(job, config);
	}

	public boolean isRunning() {
		return running;
	}

	public void stop() {
		running = false;
	}

}
